/**
 * Copyright(c) 2013 Jade Techonologies Co., Ltd.
 *
 * History:
 *   14-7-29 下午3:21 Created by dwzhang
 */
package com.jade.framework.base.comm.netty;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import com.jade.framework.base.comm.*;
import com.jade.framework.base.comm.impl.WindowResponseFuture;
import com.jade.framework.base.net.windowing.DuplicateKeyException;
import com.jade.framework.base.net.windowing.OfferTimeoutException;
import com.jade.framework.base.net.windowing.Window;
import com.jade.framework.base.net.windowing.WindowFuture;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;

/**
 * Netty传输器。
 *
 * @author dwzhang
 * @version 1.0 14-7-29 下午3:21
 */
public class NettyProtocolTransport implements ProtocolTransport
{
    protected Channel channel;
    protected Protocol protocol;
    protected Window<String, RequestMessage, ResponseMessage> sendWindow;
    protected static ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    public NettyProtocolTransport(Channel channel, Protocol protocol)
    {
        this.channel = channel;
        this.protocol = protocol;
        ProtocolConfiguration configuration = protocol.getConfiguration();
        this.sendWindow = new Window<String, RequestMessage, ResponseMessage>(configuration.getWindowSize(),
                                                                              scheduledExecutorService,
                                                                              configuration.getWindowMonitorInterval(),
                                                                              null, configuration.getName() +
                                                                                    "WindowMonitor");
    }

    public boolean isAlive()
    {
        return channel != null && channel.isActive();
    }

    /**
     * 返回远程地址。
     *
     * @return 远程地址。
     */
    public InetSocketAddress getRemoteAddress()
    {
        return (InetSocketAddress)channel.remoteAddress();
    }

    /**
     * 返回本地地址。
     *
     * @return 本地地址。
     */
    public InetSocketAddress getLocalAddress()
    {
        return (InetSocketAddress)channel.localAddress();
    }

    public void send(Object message)
            throws CommException
    {
        channel.writeAndFlush(message);
    }

    public ResponseFuture send(RequestMessage request)
            throws CommException, InterruptedException
    {
        WindowFuture<String, RequestMessage, ResponseMessage> future = null;
        ProtocolConfiguration configuration = protocol.getConfiguration();
        try {
            future = sendWindow.offer(request.getSequenceId(), request, configuration.getWindowWaitTimeout(),
                                      configuration.getRequestExpiryTimeout(), true);
            // write the pdu out & wait till its written
            ChannelFuture channelFuture = this.channel.writeAndFlush(request).await();

            // check if the write was a success
            if (!channelFuture.isSuccess()) {
                // the write failed, make sure to throw an exception
                throw new CommChannelException(channelFuture.cause().getMessage(), channelFuture.cause());
            }
            return new WindowResponseFuture(future);
        }
        catch (DuplicateKeyException e) {
            throw new DuplicatedMessageException(e.getMessage(), e);
        }
        catch (OfferTimeoutException e) {
            throw new CommTimeoutException(e.getMessage(), e);
        }
        catch (InterruptedException ie) {
            throw new CommChannelException(ie.getMessage(), ie);
        }
    }

    public ResponseMessage sendAndGetResponse(RequestMessage request, int timeout)
            throws CommException, InterruptedException
    {
        ResponseFuture future = send(request);
        return future.get(timeout);
    }

    public void onResponse(ResponseMessage response)
    {
        String sequence = response.getSequenceId();
        try {
            // see if a correlating request exists in the window
            WindowFuture<String, RequestMessage, ResponseMessage> future = sendWindow.complete(sequence, response);
            if (future != null) {
                protocol.getLog().trace("Found a future in the window for seqNum [" + sequence + "]");
            }
            else {
                protocol.getLog().warn("receive a unexpected response message:" + sequence);
            }
        }
        catch (InterruptedException e) {
            protocol.getLog().warn(
                    "Interrupted while attempting to process response and match it to a request via requesWindow: ", e);
        }
    }

    public void close()
    {
        close(5000);
    }

    public void close(long timeoutInMillis)
    {
        if (isAlive()) {
            // temporarily set to "unbinding" for now
            //this.state.set(STATE_UNBINDING);
            // make sure the channel is always closed
            if (channel.close().awaitUninterruptibly(timeoutInMillis)) {
                protocol.getLog().info("Successfully closed");
            }
            else {
                protocol.getLog().warn("Unable to cleanly close channel");
            }
        }
        //this.state.set(STATE_CLOSED);
        //release resource
        sendWindow.destroy();
        //scheduledExecutorService.shutdown();
    }
}
